package com.luboyan.CBillbill.config;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.luboyan.CBillbill.dao.user.following.UserFollowing;
import com.luboyan.CBillbill.dao.user.UserMoments;
import com.luboyan.CBillbill.service.WebSocketService;
import com.luboyan.CBillbill.service.user.following.UserFollowingService;
import com.mysql.cj.util.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import static com.luboyan.CBillbill.util.enums.UserInfoEnum.*;

/**
 * RocketMQ配置类
 * 名字服务器 NameServer
 *
 * <p>
 * NameServer是一个简单的 Topic 路由注册中心，支持 Topic、Broker 的动态注册与发现。
 * 主要包括两个功能：
 * Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
 * 路由信息管理，每个NameServer将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。
 * NameServer通常会有多个实例部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，
 * 客户端仍然可以向其它NameServer获取路由信息。
 *
 * <p>
 * 代理服务器 Broker
 * Broker主要负责消息的存储、投递和查询以及服务高可用保证。
 * NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker部署相对复杂。
 * 在 Master-Slave 架构中，Broker 分为 Master 与 Slave。一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master 与 Slave 的对应关系通过指定相同的BrokerName，
 * 不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。
 */
@Configuration
public class RocketMQConfig {

    @Value("${rocketmq.name.server.addr}")
    private String nameServerAddr;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowingService userFollowingService;

    /**
     * 实现动态消息发布的生产者
     */
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws Exception {
        //设置生产者分组
        DefaultMQProducer producer = new DefaultMQProducer(USER_MOMENTS_GROUP.getParameter());
        //设置名称服务器
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        return producer;
    }

    /**
     * 实现动态消息接收的消费者（push模式）
     */
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentConsumer() throws Exception {
        //设置消费者分组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(USER_MOMENTS_GROUP.getParameter());
        //设置名称服务器
        consumer.setNamesrvAddr(nameServerAddr);
        //设置消费者订阅的消息分组和二级分组，目前二级分组暂时设置为所有
        consumer.subscribe(TOPIC_MOMENTS.getParameter(), "*");
        //设置监听器，当生产者发布消息给mq时，消费者需要使用监听器抓取mq推送过来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * 对抓取的消息进行处理并返回
             * @param list MessageExt对消息的扩充，增加处理结果
             * @param consumeConcurrentlyContext 处理的上下文
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //以下的目的就是为了将消息写入redis中
                MessageExt msg = list.get(0);
                if (msg == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(msg.getBody()); //获取用户发送的消息体
                //将消息还原为UserMoments类型
                UserMoments userMoments = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoments.class);
                Long userId = userMoments.getUserId();
                //获取发送用户的粉丝
                List<UserFollowing> userFans = userFollowingService.getUserFans(userId);
                for (UserFollowing fans : userFans) {
                    String key = "subscribed-" + fans.getUserId();
                    String subscribedListStr = redisTemplate.opsForValue().get(key); //用户订阅的所有消息会放在redis中的一个列表里
                    List<UserMoments> subscribedList;
                    if(StringUtils.isNullOrEmpty(subscribedListStr)){
                        subscribedList = new ArrayList<>(); //说明这个fans是第一次被推送
                    }else{
                        subscribedList = JSONArray.parseArray(subscribedListStr, UserMoments.class);
                    }
                    subscribedList.add(userMoments); //将新抓取到的消息添加进去
                    redisTemplate.opsForValue().set(key,JSONObject.toJSONString(subscribedList)); //将消息写入redis
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }




    /**
     * 弹幕功能生产者
     */
    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer(DANMUS_GROUP.getParameter());
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        return producer;
    }

    /**
     * 弹幕功能消费者
     */
    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(DANMUS_GROUP.getParameter());
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.subscribe(TOPIC_DANMUS.getParameter(), "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt msg = list.get(0);
                String bodyStr = new String(msg.getBody());
                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");
                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
                if(webSocketService.getSession().isOpen()){
                    try {
                        webSocketService.sendMessage(message);
                    }catch (IOException e){
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }

}
